package personal.wang.proxyAndPekko;

import com.typesafe.config.ConfigFactory;
import org.apache.pekko.actor.*;
import personal.wang.proxy.Salable;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;

public class PekkoInvocationHandler implements InvocationHandler {
    private ActorRef actorRef ;
    public PekkoInvocationHandler(){
        ActorSystem actorSystem = ActorSystem.create("flink", ConfigFactory.load("test_client"));
        actorRef = actorSystem.actorOf(Props.create(PekkoInvocationHandler.ActorHandler.class) , "client");
    }
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        actorRef.tell(new InvocationRequest(method.getName() , args) , ActorRef.noSender());
        return null;
    }


    public static void main(String[] args) {
        Salable salable = (Salable)Proxy.newProxyInstance
                (PekkoInvocationHandler.class.getClassLoader()
                        , new Class[]{Salable.class}
                        , new PekkoInvocationHandler()
                );
        salable.sale();

    }
    public static class ActorHandler extends AbstractActor{
        @Override
        public Receive createReceive() {
            return receiveBuilder().match(InvocationRequest.class , this::messageHandler).build();
        }

        private void messageHandler(InvocationRequest request) {
            ActorSelection actorSelection = context().actorSelection("pekko://flink@127.0.0.1:17354/user/server");
            actorSelection.tell(request , getSelf());
        }
    }
}
